package com.zf.nospring.kafka.utils;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Properties;

/**
 * @author Zhou Feng
 */
public class KafkaUtils {


    /**
     * 获取一个默认的生产者对象
     */
    public static KafkaProducer<String, String> getDefaultProducer() {
        // 创建 kafka 生产者的配置对象
        Properties properties = new Properties();
        // kafka broker的地址
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        // key,value 序列化（必须）：key.serializer，value.serializer
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        // 发送成功确认机制，0、1、all 三种可选。
        // 0: 表示消息发出就算成功，不管kafka有没有收到。
        // 1: 表示只要分区的leader副本收到消息就算是发送成功，否则就失败重试。
        // all 或者 -1：表示分区的ISR（“In-Sync Replicas”，和leader保持同步的副本）中的副本都同步了leader中的该消息才算是发送成功，这是默认配置。
        properties.put(ProducerConfig.ACKS_CONFIG, "all");
        // 发送消息失败后重试几次
        properties.put(ProducerConfig.RETRIES_CONFIG, 3);
        // 批量发送的大小，单位字节，默认16KB，生产端累积16KB大小的消息后，将其一起提交到kafka。设置为0标识禁用批处理。
        // 单条消息大于bath-size时，会马上发送给kafka。
        properties.put(ProducerConfig.BATCH_SIZE_CONFIG, 16384);
        // 生产端缓冲区的大小，单位字节，默认32MB
        properties.put(ProducerConfig.BUFFER_MEMORY_CONFIG, 33554432);
        // 生产端消息在缓冲区多少毫秒后，被发送给kafka。默认为0表示立刻将消息发送给kafka,此时batch-size参数无效
        // 同时设置batch-size和 linger.ms,就是哪个条件先满足就会将消息发送出去
        properties.put(ProducerConfig.LINGER_MS_CONFIG, 50);

        // 开启幂等性，默认开启。
        // 开启后当Kafka的生产者生产消息时，会增加一个pid（每个Producer在初始化时，都会分配一个唯一的PID）和 sequence number（每个生产者对每个Topic的Partition都对应一个从0开始单调递增的SequenceNumber值），
        // 发送消息，会连着pid和sequence number一块发送，kafka接收到消息，会将消息和pid、sequence number一并保存下来，
        // 如果ack响应失败，生产者重试，再次发送消息时，Kafka会根据pid、sequence number是否需要再保存一条消息。
        // 注意：幂等性只能保证在单分区内不重复，如果Producer重启，也会导致幂等重置。
        properties.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true);

        // 创建 kafka 生产者对象
        return new KafkaProducer<>(properties);
    }

    /**
     * 获取一个默认的消费者对象
     */
    public static KafkaConsumer<String, String> getDefaultConsumer() {
        // 创建消费者的配置对象
        Properties properties = new Properties();
        // kafka broker的地址
        properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        // 配置序列化 必须
        properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        // 配置消费者组ID，任意起名，必传
        properties.put(ConsumerConfig.GROUP_ID_CONFIG, "defaultCG");

        // 当 Kafka 中没有初始偏移量或当前偏移量在服务器中不存在，该如何处理？
        // earliest：自动重置偏移量到最早的偏移量。
        // latest：默认，自动重置偏移量为最新的偏移量。
        // none：如果消费组原来的（previous）偏移量不存在，则抛异常。
        // 如果已经存在偏移量则从存在位置继续获取消息，earliest、latest、none均无效配置无效。
        properties.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest");

        // 一次 poll 拉取，获取消息的最大条数，默认是 500 条。
        properties.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 20);

        // 开启自动提交offset功能，默认开启。offset从0开始
        properties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, true);
        // 每隔多少毫秒自动提交一次 offset
        properties.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, 50);

        // 创建消费者对象
        return new KafkaConsumer<>(properties);
    }
}
